package sgt.project;

import ltq.util.*;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

public class MakeUpCountBolt extends BaseRichBolt {
    private OutputCollector collector;
    private Map<String,Integer[]> map;
    Integer []type=new Integer[]{0,0,0,0};
    Integer []type2;
    Connection conn=null;
    PreparedStatement ps = null;
    ResultSet rs = null;
    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
        this.collector=collector;
        map=new HashMap();
//        type = new Integer[]{0,0,0,0};
        type2 = new Integer[4];
        conn= DatabaseUtil.getConnection();
    }

    @Override
    public void execute(Tuple input) {
        String event_type=input.getString(1);
        String product_id=input.getString(2);

        if (map.containsKey(product_id)){
            type2=map.get(product_id);
        }else {
            type2=Arrays.copyOf(type,4);
        }


        if ("view".equals(event_type)){
            type2[0]++;
        }else if ("cart".equals(event_type)){
            type2[1]++;
        }else if ("remove_from_cart".equals(event_type)){
            type2[2]++;
        }else {
            type2[3]++;
        }
        map.put(product_id,type2);
        refreshData(product_id,type2);

        collector.emit(new Values(product_id,type2[0],type2[1],type2[2],type2[3]));
        collector.ack(input);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("product_id","view","cart","remove_from_cart","purchase"));
    }
    public void refreshData(String product_id,Integer []type){
        int view = type[0];
        int cart = type[1];
        int remove_from_cart = type[2];
        int purchase = type[3];

        try {
            ps= conn.prepareStatement("select * from product_event_type_num where product_id=?");
            ps.setObject(1,product_id);
            rs= ps.executeQuery();
            if (rs.next()){
                ps= conn.prepareStatement(
                        "update product_event_type_num " +
                                "set view=?,cart=?,remove_from_cart=?,purchase=? where product_id=?");
                ps.setObject(1,view);
                ps.setObject(2,cart);
                ps.setObject(3,remove_from_cart);
                ps.setObject(4,purchase);
                ps.setObject(5,product_id);
                ps.executeUpdate();
            }
            else {
                ps= conn.prepareStatement("insert into product_event_type_num values (?,?,?,?,?)");
                ps.setObject(1,product_id);
                ps.setObject(2,view);
                ps.setObject(3,cart);
                ps.setObject(4,remove_from_cart);
                ps.setObject(5,purchase);

                ps.executeUpdate();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

    }
}
